/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.queue.impl;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ThreadFactory;

import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityMonitor;
import org.sabayframework.channel.Channel;
import org.sabayframework.log.Log;
import org.sabayframework.queue.ExecutionQueue;

/**
 * Base class for ExecutionQueue implementations.
 * 
 * @author Alex Antonau
 * */
public abstract class BaseExecutionQueue implements ExecutionQueue {

	protected final ThreadFactory threadFactory;
	protected final List<Thread> workerThreads;
	protected volatile boolean active;
	protected volatile Throwable lastError;
	private final ThreadLocal<ActivityContextImpl<?,?>> currentContext;
		
	public BaseExecutionQueue(ThreadFactory threadFactory) {
		this.threadFactory = threadFactory;
		workerThreads=Collections.synchronizedList(new ArrayList<Thread>());
		this.currentContext = new ThreadLocal<ActivityContextImpl<?,?>>();
	}
	
	public ThreadFactory getThreadFactory() {
		return threadFactory;
	}

	public void start(int threadNumber){
		stop();
		this.active = true;
		for(int i=0;i<threadNumber;++i)
		{
			addThread();
		}
		if(Log.DEBUG) Log.debug(getClass().getName(), " started");
	}
	@Override
	public int workerCount() {
		return workerThreads.size();
	}
			
	public boolean isActive() {
		return active;
	}

	public void setActive(boolean active) {
		this.active = active;
	}

	protected abstract Runnable createWorker(int workerNo);
	
	protected int getNextWorkerNo(){
		return workerThreads.size();
	}
	void addThread(){
		final int workerNo = getNextWorkerNo();
		Runnable worker = createWorker(workerNo);
		Thread t = threadFactory.newThread(worker);
		t.setName(getClass().getName()+" - "+workerNo);
		workerThreads.add(t);
		t.start();
	}
	public void stop(){
		if(active){
			active=false;
			for(Thread t: workerThreads){
				if(t.isAlive())	t.interrupt();
			}
			for(Thread t: workerThreads){
				if(t.isAlive())
					try {
						t.join();
					} catch (InterruptedException e) {}
			}
			workerThreads.clear();
			if(Log.DEBUG) Log.debug(getClass().getName(), " stopped");
		}
	}
		
	public Throwable getLastError() {
		return lastError;
	}

	public void setLastError(Throwable lastError) {
		this.lastError = lastError;
	}
	protected abstract <I, O>  void assignToWorker(ActivityContextImpl<I, O> ctx) throws InterruptedException;
	
	@Override
	public final <I, O> ActivityMonitor<I,O> submit(Activity<I, O> a, Channel<I> input,	Channel<O> output) {
		final ActivityContextImpl<?,?> parent = currentContext.get();
		if(parent == null && !active) throw new IllegalStateException("Queue is not active");
		ActivityContextImpl<I, O> ctx = new ActivityContextImpl<I, O>(this, a, input, output, parent);
		try {
			ctx.beforeAdding(this);
			assignToWorker(ctx);
		} catch (Throwable e) {
			ctx.handleError(e);
		}
		return ctx;
	}
	
	protected final void runActivity(ActivityContextImpl<?,?> a,final boolean greedy) throws InterruptedException{
		boolean ready = true;
		final ActivityContextImpl<?,?> parent = currentContext.get();
		currentContext.set(a);
		while(ready) {
			try{
				if(Log.DEBUG) Log.debug("Enter ",  a.getActivityName());
				final long time = a.runOnce();
				if(Log.DEBUG) Log.debug("Exit ",  a.getActivityName()," - ", time);
				if(!greedy && !isEmpty()) break;
				//continue if greedy or there is no other activities
			} catch (Throwable e) {
				a.handleError(e);
				Log.error(a.getActivityName(), e);
				break;
			}
			ready = a.isDone() ? false : a.isReady();
		}
		currentContext.set(parent);
	}
	
	@Override
	public final <I, O> ActivityMonitor<I,O> execute(Activity<I, O> a, Channel<I> input, Channel<O> output) {
		final ActivityContextImpl<?,?> parent = currentContext.get();
		if(parent==null && !active) throw new IllegalStateException("Queue is not active");
		ActivityContextImpl<I, O> ctx = new ActivityContextImpl<I, O>(this, a, input, output, parent);
		try {
			ctx.beforeAdding(this);
			if(ctx.isReady()) {					
				runActivity(ctx, true);
				if(!ctx.isDone()) {
					assignToWorker(ctx);
				}
				else {
					ctx.afterRemoving(this);
				}
			}else{
				assignToWorker(ctx);
			}
		} catch (Throwable e) {
			ctx.handleError(e);
		}
		return ctx;
	}

}
